package com.wunong.smart.bigdata.flink.job;

import com.wunong.smart.bigdata.flink.data.AnalysisData;
import com.wunong.smart.bigdata.flink.data.SourceData;
import com.wunong.smart.bigdata.flink.function.ExcludeValueFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author created by zealot.zt
 */
public class ObjectConnectJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment()
                .setParallelism(1);
        // 需要排除的流
        KeyedStream<SourceData, String> sourceStream = env.fromElements(
                        SourceData.create("DROP"),
                        SourceData.create("IGNORE"))
                .keyBy(SourceData::getId);
        // 元数据流
        KeyedStream<AnalysisData, String> dataStream = env.fromElements(
                        AnalysisData.create("Apache"),
                        AnalysisData.create("DROP"),
                        AnalysisData.create("Flink"),
                        AnalysisData.create("IGNORE"))
                .keyBy(AnalysisData::getSourceId);

        // connected stream 实现流的关联
        sourceStream.connect(dataStream)
                .flatMap(new ExcludeValueFunction())
                .print("ConnectResult");

        env.execute(ObjectConnectJob.class.getName());
    }
}